package com.im.dispatcher;

import com.alibaba.fastjson.JSONObject;
import com.im.common.ImConstant;
import com.im.common.Request;
import com.im.common.Response;
import com.im.common.SessionInfo;
import com.im.protocol.GroupMessageResponseProto;
import com.im.protocol.MessageResponseProto;
import com.im.protocol.PushGroupMessageRequestProto;
import com.im.protocol.PushMessageRequestProto;
import io.netty.channel.socket.SocketChannel;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import redis.clients.jedis.Jedis;

import java.util.Arrays;

/**
 * @Created leijun
 * @Date 2021/10/16 10:25 上午
 */
public class DispatcherMessageThread extends Thread{

    private KafkaConsumer<String, String> consumer;

    @Override
    public void run() {
        KafkaManager kafkaManager = KafkaManager.getInstance();
        consumer = kafkaManager.getConsumer();
        consumer.subscribe(Arrays.asList("single_message_response", "group_message_response", "push_message", "push_group_message"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                JSONObject message = JSONObject.parseObject(record.value());
                String topic = record.topic();
                if (topic.equals("single_message_response")) {
                    handleSingleMessageResponse(message);
                } else if (topic.equals("group_message_response")) {
                    handleGroupMessageResponse(message);
                } else if (topic.equals("push_message")) {
                    handlePushSingleMessage(message);
                } else if (topic.equals("push_group_message")) {
                    handlePushGroupMessage(message);
                }
            }
        }
    }

    /**
     * 处理单聊消息响应消息
     * @param message
     */
    private static void handleSingleMessageResponse(JSONObject message) {
        MessageResponseProto.MessageResponse.Builder builder =
                MessageResponseProto.MessageResponse.newBuilder();
        builder.setSenderId(message.getLong("senderId"))
                .setMessageId(message.getLong("messageId"))
                .setTimestamp(message.getLong("timestamp"));

        String senderSessionKey = "session_"+message.getLong("senderId");
        Jedis jedis = JedisManager.getJedis();
        SessionInfo sessionInfo = JSONObject.parseObject(jedis.get(senderSessionKey), SessionInfo.class);
        GatewayInstanceManager gatewayInstanceManager = GatewayInstanceManager.getInstance();
        SocketChannel senderSocketChannel = gatewayInstanceManager.getGatewayInstance(sessionInfo.getChannelId());
        Response response = new Response(ImConstant.APP_SDK_VERSION, ImConstant.REQUEST_TYPE_SEND_MESSAGE,
                ImConstant.SEQUENCE_DEFAULT, builder.build().toByteArray());
        senderSocketChannel.writeAndFlush(response.getBuf());
    }

    /**
     * 处理群聊消息响应消息
     * @param message
     */
    private static void handleGroupMessageResponse(JSONObject message) {
        GroupMessageResponseProto.GroupMessageResponse.Builder builder =
                GroupMessageResponseProto.GroupMessageResponse.newBuilder();
        builder.setSenderId(message.getLong("senderId"))
                .setGroupId(message.getLong("groupId"))
                .setMessageId(message.getLong("messageId"))
                .setTimestamp(message.getLong("timestamp"));

        String senderSessionKey = "session_"+message.getLong("senderId");
        Jedis jedis = JedisManager.getJedis();
        SessionInfo sessionInfo = JSONObject.parseObject(jedis.get(senderSessionKey), SessionInfo.class);
        GatewayInstanceManager gatewayInstanceManager = GatewayInstanceManager.getInstance();
        SocketChannel senderSocketChannel = gatewayInstanceManager.getGatewayInstance(sessionInfo.getChannelId());
        Response response = new Response(ImConstant.APP_SDK_VERSION, ImConstant.REQUEST_TYPE_GROUP_MESSAGE,
                ImConstant.SEQUENCE_DEFAULT, builder.build().toByteArray());
        senderSocketChannel.writeAndFlush(response.getBuf());
    }

    /**
     * 处理推送单聊消息
     * @param message
     */
    private static void handlePushSingleMessage(JSONObject message) {
        PushMessageRequestProto.PushMessageRequest.Builder builder =
                PushMessageRequestProto.PushMessageRequest.newBuilder();
        builder.setSenderId(message.getLong("senderId"))
                .setReceiverId(message.getLong("receiverId"))
                .setMessageId(message.getLong("messageId"))
                .setTimestamp(message.getLong("timestamp"))
                .setContent(message.getString("content"));
        Request request = new Request(ImConstant.APP_SDK_VERSION, ImConstant.REQUEST_TYPE_PUSH_MESSAGE,
                ImConstant.SEQUENCE_DEFAULT, builder.build().toByteArray());
        String receiverSessionKey = "session_"+message.getLong("receiverId");
        Jedis jedis = JedisManager.getJedis();
        SessionInfo sessionInfo = JSONObject.parseObject(jedis.get(receiverSessionKey), SessionInfo.class);
        GatewayInstanceManager gatewayInstanceManager = GatewayInstanceManager.getInstance();
        SocketChannel receiverSocketChannel = gatewayInstanceManager.getGatewayInstance(sessionInfo.getChannelId());
        receiverSocketChannel.writeAndFlush(request.getBuf());
    }

    /**
     * 处理推送群聊消息
     * @param message
     */
    private static void handlePushGroupMessage(JSONObject message) {
        PushGroupMessageRequestProto.PushGroupMessageRequest.Builder builder =
                PushGroupMessageRequestProto.PushGroupMessageRequest.newBuilder();
        builder.setSenderId(message.getLong("senderId"))
                .setReceiverId(message.getLong("receiverId"))
                .setGroupId(message.getLong("groupId"))
                .setMessageId(message.getLong("messageId"))
                .setTimestamp(message.getLong("timestamp"))
                .setContent(message.getString("content"));
        Request request = new Request(ImConstant.APP_SDK_VERSION, ImConstant.REQUEST_TYPE_PUSH_GROUP_MESSAGE,
                ImConstant.SEQUENCE_DEFAULT, builder.build().toByteArray());
        String receiverSessionKey = "session_"+message.getLong("receiverId");
        Jedis jedis = JedisManager.getJedis();
        SessionInfo sessionInfo = JSONObject.parseObject(jedis.get(receiverSessionKey), SessionInfo.class);
        GatewayInstanceManager gatewayInstanceManager = GatewayInstanceManager.getInstance();
        SocketChannel receiverSocketChannel = gatewayInstanceManager.getGatewayInstance(sessionInfo.getChannelId());
        receiverSocketChannel.writeAndFlush(request.getBuf());
    }
}
